package com.example.reactor.rsocket.config;

import com.example.reactor.rsocket.config.entity.TimeSeriesInfo;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.stereotype.Component;
import org.springframework.util.MimeTypeUtils;
import reactor.core.publisher.Flux;

import java.util.List;
import java.util.stream.IntStream;

@Component
@Slf4j
public class ReactiveNodeClient {
    private final RSocketRequester rSocketRequester;

    public ReactiveNodeClient(RSocketRequester.Builder builder,
                              @Value("${futurmaster.masterdata.rsocket.host:localhost}") String host,
                              @Value("${futurmaster.masterdata.rsocket.port:7000}") int port) {
        this.rSocketRequester = builder.tcp(host, port);
    }

    public Flux<String> multiParam(String user, Flux<String> list) {
        TimeSeriesInfo timeSeriesInfo1 = getTimeSeriesInfo();

        log.info("before start multiParam");

        Flux<String> stringFlux = rSocketRequester
                .route("multiParamFlux")
                .metadata(timeSeriesInfo1, MimeTypeUtils.parseMimeType("application/json"))
                .data(list)
                .retrieveFlux(String.class)
                .contextWrite(context -> context
                        .put("user", user)
                        .put("tenantId", "myTenantId")
                );
        log.info("after start multiParam");
        return stringFlux;
    }

    public Flux<String> multiParam2(String user, Flux<String> list) {

        log.info("before start multiParam");
        Flux<String> stringFlux = rSocketRequester
                .route("multiParamFlux2")
                .metadata(user, MimeTypeUtils.TEXT_PLAIN)
                .data(list)
                .retrieveFlux(String.class);
        log.info("after start multiParam");
        return stringFlux;
    }

    private static TimeSeriesInfo getTimeSeriesInfo() {
        TimeSeriesInfo timeSeriesInfo1 = new TimeSeriesInfo();
        timeSeriesInfo1.setId(100);
        timeSeriesInfo1.setVersion(5000);
        String repeat = "a".repeat(1024);
        List<String> payload = IntStream.range(0, 1024 * 15)
                .boxed()
                .map(item -> repeat)
                .toList();
        timeSeriesInfo1.setPayload(payload);
        return timeSeriesInfo1;
    }

    public Flux<String> multiParam3(String user, Flux<String> list) {
        ByteBuf customMetadata = ByteBufAllocator.DEFAULT.buffer();

        TimeSeriesInfo timeSeriesInfo1 = getTimeSeriesInfo();

        log.info("before start multiParam");

        Flux<String> stringFlux = rSocketRequester
                .route("multiParamFlux")
                .metadata(timeSeriesInfo1, MimeTypeUtils.parseMimeType("application/custom-metadata"))
                .data(list)
                .retrieveFlux(String.class)
                .contextWrite(context -> context
                        .put("user", user)
                        .put("tenantId", "myTenantId")
                );
        log.info("after start multiParam");
        return stringFlux;
    }
}
